package org.robot.scm.platform.server.common;

import static org.robot.scm.platform.server.LockKey.LAST_KEY;
import static org.robot.scm.platform.server.LockKey.LOCK_KEY;

import cn.hutool.core.io.FileUtil;
import com.google.common.collect.Lists;
import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.util.List;
import java.util.Objects;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.robot.scm.infra.LocalCacheHelper;
import org.robot.scm.infra.RedisCacheHelper;
import org.robot.scm.platform.registry.HikariRegistry;
import org.robot.scm.platform.registry.SessionRegistry;
import org.robot.scm.platform.registry.TelegramRegistry;
import org.robot.scm.pojos.config.BaseConfig;
import org.robot.scm.pojos.enums.PlatformEnum;
import org.robot.scm.pojos.platform.MessageBaseNotify;
import org.robot.scm.pojos.platform.server.ExecuteParam;
import org.robot.scm.pojos.platform.server.SshMySql;
import org.robot.scm.utils.DateTool;
import org.robot.scm.utils.JsonHelper;
import org.robot.scm.utils.SendNoticeHelper;
import org.robot.scm.utils.StreamUtils;

/**
 * @Author : Awn create on 2024/5/7 19:45
 */
@Slf4j
public class DefaultRunner {

    public void start(BaseConfig config, ExecuteParam param) {
        try {
            StopWatch stopWatch = StopWatch.createStarted();
            RunnerLogHelper.clearLog(param.getFileKey());
            boolean isAllSucc = this.startTask(config, param);
            RunnerLogHelper.appendEndLog(param.getFileKey(), "");
            RunnerLogHelper.appendEndLog(param.getFileKey(), "release lock");
            RunnerLogHelper.appendEndLog(param.getFileKey(), "execute complete, cost: " + stopWatch.getTime() + "ms");

            //发送消息通知到群, 授权码在哪个群就通知哪个群
            if (Objects.nonNull(param.getIsShellExec()) && param.getIsShellExec()) {
                return;
            }
            String joined = param.initParams().size() == config.getCurrentPlatform().getSshMysqlDetails().size() ? "所有环境" : String.join(",", param.initParams());
            String platforms = joined.replaceAll("mec", "").replaceAll("sec", "").replaceAll("mvd", "").replaceAll("svd", "");
            String isSuccString = isAllSucc ? "✅ 执行成功" : "❌ 执行异常";
            String template = "【工单】%s\n【文件】%s\n【行数】处理%s条记录\n【环境】%s\n【异常】%s\n【用时】%s";

            String fileName = RedisCacheHelper.get(param.getFileKey());
            if (StringUtils.isBlank(fileName)) {
                fileName = TelegramRegistry.likeSqlFile.get();
            }
            String content = String.format(template, maskString(param.getFileKey()), maskString(StringUtils.isBlank(fileName) ? "非文件脚本" : fileName), param.getFileLine(), platforms, isSuccString, DateTool.millToHms(stopWatch.getTime()));
            SendNoticeHelper.sendNotice(config, MessageBaseNotify.builder().message(content).build(), PlatformEnum.VERIFY);
        } finally {
            RunnerLogHelper.appendEndLog(param.getFileKey(), "log delete after 5 minutes");
            removeAfterMinute(param.getFileKey());
            RedisCacheHelper.remove(LOCK_KEY);
            // 任务执行完成记录当前的key保持120秒，用于app后台断连后未及时查看导致的查询时间差
            RedisCacheHelper.put(LAST_KEY, param.getFileKey(), 600);
            // 清理sqlLike
            TelegramRegistry.clearSqlLike();
        }
    }


    public static String maskString(String input) {
        if (StringUtils.isBlank(input)) {
            return StringUtils.EMPTY;
        }
        if (input.length() <= 12) {
            return input;
        }
        int length = input.length();
        int prefixLength = 5;
        int suffixLength = Math.min(5, length - 12);
        return input.substring(0, prefixLength) + "****" + input.substring(length - suffixLength);
    }

    /**
     * 10分钟后移除日志
     */
    private void removeAfterMinute(String fileKey) {
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                LocalCacheHelper.remove(fileKey);
                // FileUtil.del(new File("assets/telegram/" + fileKey.concat(".sql")));
            }
        }, 600 * 1000L);
    }

    private boolean startTask(BaseConfig config, ExecuteParam param) {
        List<String> params = param.initParams();
        RunnerLogHelper.appendStartLog(param.getFileKey(), param.getFileKey());
        RunnerLogHelper.appendStartLog(param.getFileKey(), JsonHelper.toJSONString(params));
        if (param.getFileLine() <= 0) {
            param.setFileLine(FileUtil.getTotalLines(FileUtil.newFile("assets/telegram/" + param.getFileKey().concat(".sql"))));
            RunnerLogHelper.appendStartLog(param.getFileKey(), "read script: " + param.getFileLine() + " lines");
        }

        // 如果没有线程池支持, 系统会根据cpu核数创建线程池，在目标服务器EC2上同时只能跑3个
        AtomicBoolean isAllSucc = new AtomicBoolean(true);
        ExecutorService executor = Executors.newFixedThreadPool(params.size());
        List<CompletableFuture<Void>> futures = StreamUtils.convert(params, aws -> CompletableFuture.runAsync(() -> {
            try {
                RunnerLogHelper.appendLog(param.getFileKey(), aws, "");
                initRunner(config.getCurrentPlatform().getSshMysql(aws), param);
            } catch (Exception e) {
                isAllSucc.set(false);
                log.info("execute error", e);
                RunnerLogHelper.appendLog(param.getFileKey(), aws, aws + " error: " + e.getMessage());
            }
        }, executor));
        futures.forEach(CompletableFuture::join);
        executor.shutdown();
        return isAllSucc.get();
    }

    private void initRunner(SshMySql sshMysql, ExecuteParam param) {
        Session session = SessionRegistry.getSessionMap().get(sshMysql.getName());
        // 如果是清理任务或者sql则上传到服务器执行
        if (Objects.nonNull(param.getIsCleanTask()) && param.getIsCleanTask()) {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "open shell for " + sshMysql.getName());
            initRemoteCommand(sshMysql, session, param);
            return;
        }
        if (Objects.nonNull(param.getIsShellExec()) && param.getIsShellExec()) {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "open shell for " + sshMysql.getName());
            execRemoteCommand(sshMysql, session, param);
            return;
        }
        RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "open session for " + sshMysql.getName());

        // 16套环境因为binlog问题未解决，无法使用mybatis-scriptrunner运行，只能上传脚本shell执行
        if (Lists.newArrayList("mec16", "mec19", "mec23").contains(sshMysql.getName())) {
            initRemoteCommand(sshMysql, session, param);
            return;
        }
        initCommonCommand(sshMysql, param);
    }

    /**
     * 普通脚本执行，通过ScmScriptRunner
     */
    private void initCommonCommand(SshMySql sshMysql, ExecuteParam param) {
        Path path = Paths.get("assets/telegram/" + param.getFileKey().concat(".sql"));
        try (Connection connection = HikariRegistry.getHikariMap().get(sshMysql.getName()).getConnection()) {
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(Files.newInputStream(path), StandardCharsets.UTF_8))) {
                // 创建ScriptRunner对象
                ScmScriptRunner scriptRunner = new ScmScriptRunner(connection);

                // 删除sql换行
                scriptRunner.setRemoveCRs(Boolean.TRUE);

                // 执行错误日志记录
                PrintWriter errorPrintWriter = new PrintWriter(new StringWriter());
                BackPrintWriter errorBackWriter = new BackPrintWriter(errorPrintWriter, data -> RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), data));
                scriptRunner.setErrorLogWriter(errorBackWriter);

                // 输出sql与结果记录
                PrintWriter logPrintWriter = new PrintWriter(new StringWriter());
                BackPrintWriter logBackWriter = new BackPrintWriter(logPrintWriter, data -> RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), data));
                scriptRunner.setLogWriter(logBackWriter);

                //遇到错误终止执行
                scriptRunner.setStopOnError(Boolean.TRUE);

                // 执行SQL脚本, 这里有个hook，判断是否需要终止，判断依据是当前的locker是否存在
                scriptRunner.runScript(reader, param.getFileLine());

                //关闭流
                errorPrintWriter.close();
                errorBackWriter.close();
                logPrintWriter.close();
                logBackWriter.close();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "close session for " + sshMysql.getName());
        }
    }

    private void initRemoteCommand(SshMySql sshMysql, Session session, ExecuteParam param) {
        try {
            String remoteDir = "/home/ec2-user/clean-script/";
            String localFilePath = "assets/telegram/" + param.getFileKey().concat(".sql");
            String remoteScriptPath = remoteDir + param.getFileKey().concat(".sql");
            List<String> scriptLines = Files.readAllLines(Paths.get(localFilePath));

            // 1. 先创建目录、文件
            this.executeCommand(session, "mkdir -p " + remoteDir + "\n" + "touch " + remoteScriptPath + "\n");

            // 2. 分块写入SQL文件内容，优化根据行长度动态调整分块大小
            int chunkSize = calculateOptimalChunkSize(scriptLines);
            for (int i = 0; i < scriptLines.size(); i += chunkSize) {
                int end = Math.min(i + chunkSize, scriptLines.size());
                List<String> chunk = scriptLines.subList(i, end);
                String command = buildUploadCommand(chunk, remoteScriptPath, i == 0);
                executeCommand(session, command);
            }

            // 3. 执行脚本
            StringBuilder commandBuilder = new StringBuilder();
            commandBuilder.append("cat ").append(remoteScriptPath).append(" | ");
            commandBuilder.append("/usr/local/mysql/bin/mysql -u").append(sshMysql.getMysqlUser());
            commandBuilder.append(" -h127.0.0.1 -P3306 -p'").append(sshMysql.getMysqlPass()).append("' --verbose 2>&1");

            ChannelExec channel = (ChannelExec) session.openChannel("exec");
            channel.setCommand(commandBuilder.toString());

            channel.setInputStream(null);
            channel.setPty(true);
            channel.connect();

            // 改进的流读取
            readStreams(channel, param.getFileKey(), sshMysql.getName());

            channel.disconnect();
        } catch (Exception e) {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "error exec: " + e.getMessage());
        } finally {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "close session for " + sshMysql.getName());
        }
    }

    private String buildUploadCommand(List<String> chunk, String remotePath, boolean isFirst) {
        String action = isFirst ? ">" : ">>";
        String content = String.join("\n", chunk);
        return String.format("cat %s %s << 'EOF'\n%s\nEOF", action, remotePath, content);
    }

    /**
     * 基于平均行长度计算合适的分块大小
     */
    private int calculateOptimalChunkSize(List<String> lines) {
        double avgLineLength = lines.stream()
                .mapToInt(String::length)
                .average()
                .orElse(100);
        // 目标每块约 32KB，避免 SSH 命令过长限制
        int targetChunkSize = (int) (32768 / avgLineLength);
        return Math.max(10, Math.min(200, targetChunkSize)); // 限制在 10-200 行之间
    }

    /**
     * 执行单个命令并等待完成
     */
    private void executeCommand(Session session, String command) throws Exception {
        ChannelExec channel = (ChannelExec) session.openChannel("exec");
        channel.setCommand(command);
        channel.connect();
        // 等待命令执行完成
        while (!channel.isClosed()) {
            Thread.sleep(100);
        }
        // 检查命令执行状态
        if (channel.getExitStatus() != 0) {
            throw new RuntimeException("command failed: " + command);
        }
        channel.disconnect();
    }

    private void readStreams(ChannelExec channel, String fileKey, String sshName) {
        try {
            InputStream in = channel.getInputStream();
            InputStream err = channel.getErrStream();

            // 使用线程分别读取标准输出和错误输出
            Thread outThread = new Thread(() -> readStream(in, fileKey, sshName));
            Thread errThread = new Thread(() -> readStream(err, fileKey, sshName));

            outThread.start();
            errThread.start();

            // 等待通道关闭
            while (!channel.isClosed()) {
                Thread.sleep(100);
            }

            // 等待线程结束
            outThread.join(1000);
            errThread.join(1000);
        } catch (Exception e) {
            RunnerLogHelper.appendLog(fileKey, sshName, "error reading streams: " + e.getMessage());
        }
    }

    private void readStream(InputStream stream, String fileKey, String sshName) {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (!line.trim().isEmpty() && validLineLog(line)) {
                    RunnerLogHelper.appendLog(fileKey, sshName, line);
                }
            }
        } catch (IOException e) {
            RunnerLogHelper.appendLog(fileKey, sshName, "stream read error: " + e.getMessage());
        }
    }

    /**
     * 执行远程shell脚本
     */
    private void execRemoteCommand(SshMySql sshMysql, Session session, ExecuteParam param) {
        try {
            //打开shell并执行命令
            Channel channel = session.openChannel("shell");
            channel.connect();
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "shell connect success");

            // shell命令写入
            try (PrintStream stream = new PrintStream(channel.getOutputStream(), true)) {
                // 支持多行命令脚本
                for (String command : param.getScript().split("\\r?\\n")) {
                    stream.println(command);
                }
                stream.println("exit"); // 添加结束终端命令
            }

            // 读取shell输出
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(channel.getInputStream(), StandardCharsets.UTF_8))) {
                String line;
                while ((line = removeAnsiEscape(reader.readLine())) != null) {
                    if (!validLineShell(line)) {
                        continue;
                    }
                    RunnerLogHelper.shellLog(param.getFileKey(), sshMysql.getName(), line);
                }
            }
            channel.disconnect();
        } catch (JSchException | IOException e) {
            throw new RuntimeException(e);
        } finally {
            RunnerLogHelper.appendLog(param.getFileKey(), sshMysql.getName(), "close shell for " + sshMysql.getName());
        }
    }

    /**
     * 正则表达式来移除ANSI转义序列
     */
    private static String removeAnsiEscape(String line) {
        if (StringUtils.isBlank(line)) {
            return line;
        }
        return line.replaceAll("(?i)\\x1B\\[[0-9;]*[a-zA-Z]", "");
    }

    /**
     * 规律一些日志规则
     */
    private boolean validLineLog(String line) {
        return !StringUtils.isBlank(line)
                && !StringUtils.contains(line, "--verbose")
                && !StringUtils.equals(line.trim(), "--------------")
                && !StringUtils.contains(line.trim(), "Using a password on the command");
    }

    private boolean validLineShell(String line) {
        return !StringUtils.isBlank(line) && !StringUtils.equals(line.trim(), "--------------") &&
                !StringUtils.contains(line, "生产编号") &&
                !StringUtils.contains(line, "* * *") &&
                !StringUtils.contains(line, "AWS") &&
                !StringUtils.contains(line, "cannot modify limit") &&
                !StringUtils.contains(line, "apply all updates") &&
                !StringUtils.contains(line, "needed for security");
    }

}
